package com.duowan.realtime.scheduled.batch.writer;

import com.duowan.common.util.DateConvertUtils;
import com.duowan.realtime.dao.redis.RedisDao;
import com.google.common.collect.Lists;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.item.ItemWriter;
import redis.clients.jedis.Pipeline;

import java.util.Date;
import java.util.List;
import java.util.Map;

/**
 * @author tangsicheng
 * @version 1.0
 * @since 1.0
 */
public class RedisWriter implements ItemWriter<List<Map>> {

    private static final Logger logger = LoggerFactory.getLogger(RedisWriter.class);
    private RedisDao redisDao;


    private String keySetNameFormat;


    private List<Map<String, String>> merge(List<? extends List<Map>> items) {
        List<Map<String, String>> result = Lists.newLinkedList();
        for (List list : items) {
            result.addAll(list);
        }
        return result;
    }


    @Override
    public void write(List<? extends List<Map>> items) throws Exception {
        try {
            final List<Map<String, String>> item = merge(items);
            Pipeline p = redisDao.getRedisTemplate().pipelined();
            for (Map<String, String> redisKV : item) {
                String redisKey = redisKV.get("redisKey");
                String redisValue = redisKV.get("redisValue");
                if (StringUtils.isNotBlank(redisKey)) {
                    p.set(redisKey, redisValue);
                    if (StringUtils.isNotBlank(keySetNameFormat)) {
                        p.sadd(getKeySetName(), redisKey);
                    }
                }
            }
            p.sync();
        } catch (Throwable e) {
            //如果这里不吃掉异常，整个batch job会停掉，除非在batch job加入retry-limit机制。
            logger.error(e.getMessage(), e);
        }
    }

    private String getKeySetName() {
        return DateConvertUtils.format(new Date(), keySetNameFormat);
    }

    public void setRedisDao(RedisDao redisDao) {
        this.redisDao = redisDao;
    }


    public void setKeySetNameFormat(String keySetnameFormat) {
        this.keySetNameFormat = keySetnameFormat;
    }
}
